package com.yu.rabbitmq.four;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;
import com.yu.rabbitmq.utils.RabbitMqUnits;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.TimeoutException;

/**
 * 发布确认三种模式
 */
public class confirm {

    public static void main(String[] args) throws Exception{
        //1.单个发版
        //singleSendMessage();
        //2.批量发版
        //multipleSendMessage();
        //3.异步批量发送
        AsyncSendMessage();

    }
    //单个发布
    public static void singleSendMessage() throws IOException, TimeoutException, InterruptedException {
        Channel channel = RabbitMqUnits.getChannel();
        channel.confirmSelect();//开启发布确认模式
        String queueName = "Single_Queue";
        channel.queueDeclare(queueName,false,false,false,null);
        long start = System.currentTimeMillis();
        for (int i = 0; i < 1000; i++) {
            channel.basicPublish("",queueName,null,(i+"").getBytes(StandardCharsets.UTF_8));
            boolean flag = channel.waitForConfirms();
            if (flag)
                System.out.println("消息发送成功");
        }
        long time = System.currentTimeMillis()-start;
        System.out.println("共花费:"+time+"ms");
        channel.close();
    }

    //批量确认
    public static void multipleSendMessage()throws IOException, TimeoutException, InterruptedException{
        Channel channel = RabbitMqUnits.getChannel();
        channel.confirmSelect();//开启发布确认模式
        String queueName = "Multiple_Queue";
        channel.queueDeclare(queueName,false,false,false,null);
        int batchSize = 100; //批量确认大小，没发100条就确认
        long start = System.currentTimeMillis();
        for (int i = 0; i < 1000; i++) {
            channel.basicPublish("",queueName,null,(i+"").getBytes(StandardCharsets.UTF_8));
            if (i%batchSize==0){
                channel.waitForConfirms(); //确定消息
            }
        }
        long time = System.currentTimeMillis()-start;
        System.out.println("批量共花费:"+time+"ms");
        channel.close();
    }

    //异步发送
    public static void AsyncSendMessage()throws IOException, TimeoutException, InterruptedException{

        Channel channel = RabbitMqUnits.getChannel();
        channel.confirmSelect();//开启发布确认模式
        String queueName = "Async_Queue";
        channel.queueDeclare(queueName,false,false,false,null);
        long start = System.currentTimeMillis();
        ConcurrentSkipListMap<Long,String> confirmMap = new ConcurrentSkipListMap<>();
        //准备一个消息监听器，来确定消息的成功 和 失败
        ConfirmCallback confirmCallback  = (deliveryTag, multiple)->{
            //监听成功的消息
            if (multiple){
                ConcurrentNavigableMap<Long, String> longStringConcurrentNavigableMap =
                        confirmMap.headMap(deliveryTag, true);
                longStringConcurrentNavigableMap.clear();
            }else {
                confirmMap.remove(deliveryTag);
            }
            System.out.println("已确定的消息"+deliveryTag);
        };
        ConfirmCallback nackCallback  = (deliveryTag, multiple)->{
            //监听失败的消息
            String message = confirmMap.get(deliveryTag);
            System.out.println("发布的消息:"+message+"未被确定，未确定的序号是"+deliveryTag);
        };
        channel.addConfirmListener(confirmCallback,nackCallback); //异步监听器
        for (int i = 0; i < 1000; i++) {
            confirmMap.put(channel.getNextPublishSeqNo(),i+""); //将所有消息存放到一个线程安全的map中
            channel.basicPublish("",queueName,null,(i+"").getBytes(StandardCharsets.UTF_8));
        }

        long time = System.currentTimeMillis()-start;
        System.out.println("异步共花费:"+time+"ms");
        channel.close();
    }

}
